/**
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */

#include <cstdio>
#include <iostream>
#include <fstream>
#include "Bruinbase.h"
#include "SqlEngine.h"
#include <stdlib.h>
#include <string>

using namespace std;

// external functions and variables for load file and sql command parsing 
extern FILE* sqlin;
int sqlparse(void);


RC SqlEngine::run(FILE* commandline)
{
  fprintf(stdout, "Bruinbase> ");

  // set the command line input and start parsing user input
  sqlin = commandline;
  sqlparse();  // sqlparse() is defined in SqlParser.tab.c generated from
               // SqlParser.y by bison (bison is GNU equivalent of yacc)

  return 0;
}

void printCond(const vector<SelCond> c)
{
    for(unsigned i = 0; i< c.size(); i++){
        cout << c[i].attr << " " <<c[i].comp<<endl;
    
    }
    
}
/**
   * Takes in the defined condition list, and filter it
   * @param cond[IN] the input stream to get user commands
   * @param newCond[OUT] The filtered condition list
   * @param starTKey[OUT] the the start key 
   * @param endKey[OUT] the end key
   * @return true if no conflist, false otherwise
   */
bool filterSelCond(const vector<SelCond>& conds, vector<SelCond>& newConds, int& startKey, int& endKey, bool& containValueinWhere)
{
    /*
     * The idea here is that I want to filter out the select conditions
     * Since if there are no conditions, the program should traverse from the tuple with the
     * SMALEST key to tuple with the LARGEST key. However, if there is a condition that sets
     * > or >= on the key, the lower bound would be set to that condition. This is also true
     * for condition < or <=. If the condition is =, then the lower bound/upper bound are
     * the same. Also, if the condition conflict with each other (e.g. >4 AND <3) then
     * this function will return false
     * 
     */
    
    //smallest and largest possible value for signed 4 bytes int type
    startKey = -2147483648;
    endKey   = 2147483647;
    newConds.clear();
    containValueinWhere=false;

    for(unsigned i = 0; i< conds.size(); i++){
        
        //if the condition is on value, automatically push it to the new condition list
        if(conds[i].attr == 2){
            
            containValueinWhere = true;
            newConds.push_back(conds[i]);
        }else{

            //check conditions
            switch (conds[i].comp) {
                case SelCond::EQ:
                    startKey = endKey = atoi(conds[i].value);
	            break;
                case SelCond::NE:
                    newConds.push_back(conds[i]);
	            break;
                case SelCond::GT:
                    //if its( >4), we actually want keys that are greater than or equal to 5
                    if(startKey <= atoi(conds[i].value))
                        startKey = atoi(conds[i].value) +1;
	            break;
                case SelCond::GE:
                    if(startKey < atoi(conds[i].value))
                        startKey = atoi(conds[i].value);
	            break;
                case SelCond::LT:
                    //if its( <4), we actually want keys that are smaller than or equal to 3
                    if(endKey >= atoi(conds[i].value))
                        endKey = atoi(conds[i].value)-1;

                    break;
                
                case SelCond::LE:
                    if(endKey > atoi(conds[i].value))
                        endKey = atoi(conds[i].value);
                    break;
            }
        }
        
        //check for conflict
        if(startKey > endKey)
            return false;
    }
    
    return true;
}

RC SqlEngine::select(int attr, const string& table, const vector<SelCond>& cond)
{
  RecordFile rf;   // RecordFile containing the table
  RecordId   rid;  // record cursor for table scanning
  BTreeIndex tree; // B+tree index file

  RC     rc;
  int    key;     
  string value;
  int    count;
  int    diff;

  // open the table file
  if ((rc = rf.open(table + ".tbl", 'r')) < 0) {
    fprintf(stderr, "Error: table %s does not exist\n", table.c_str());
    return rc;
  }

  // scan the table file from the beginning
  rid.pid = rid.sid = 0;
  count = 0;
  
  //Check if we need to use the index file
  bool useIndex = true;
  rc = tree.open(table +".idx", 'r');
  //case: no index file
  if( rc < 0){
      cout <<"no index file"<<endl;
      useIndex = false;
  //case: no WHERE conditions and we are selecting value or *
  }else if(cond.size() == 0){
      if(attr == 2 || attr ==3)
          useIndex = false;
  }else{
      bool hasRangeComp = false;
      for(unsigned i = 0; i< cond.size(); i++){
          //case: at least one condition is not <>(not equal) on key 
          if(cond[i].comp != SelCond::NE & cond[i].attr == 1){
              hasRangeComp = true;
              break;
          }   
      }
      useIndex = hasRangeComp;
  }

  
  //if we dont need index, just do the good old table scan
  if(useIndex != true)
  {
      while (rid < rf.endRid()) {
          if ((rc = rf.read(rid, key, value)) < 0) {
              fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
              goto exit_select;
          }

          // check the conditions on the tuple
          for (unsigned i = 0; i < cond.size(); i++) {
              // compute the difference between the tuple value and the condition value
              switch (cond[i].attr) {
              case 1:
                  diff = key - atoi(cond[i].value);
	          break;
              case 2:
	          diff = strcmp(value.c_str(), cond[i].value);
	          break;
              }

              // skip the tuple if any condition is not met
              switch (cond[i].comp) {
                  case SelCond::EQ:
                      if (diff != 0) goto next_tuple;
	              break;
                  case SelCond::NE:
	              if (diff == 0) goto next_tuple;
	              break;
                  case SelCond::GT:
	              if (diff <= 0) goto next_tuple;
	              break;
                  case SelCond::LT:
	              if (diff >= 0) goto next_tuple;
	              break;
                  case SelCond::GE:
	              if (diff < 0) goto next_tuple;
	              break;
                  case SelCond::LE:
    	              if (diff > 0) goto next_tuple;
    	              break;
              }
          }

          // the condition is met for the tuple. 
          // increase matching tuple counter
          count++;

          // print the tuple 
          switch (attr) {
              case 1:  // SELECT key
                  fprintf(stdout, "%d\n", key);
                  break;
              case 2:  // SELECT value
                  fprintf(stdout, "%s\n", value.c_str());
                  break;
              case 3:  // SELECT *
                  fprintf(stdout, "%d '%s'\n", key, value.c_str());
                  break;
          }

          // move to the next tuple
      next_tuple:
          ++rid;
    } 
      //DEBUG
      //fprintf(stderr, "This doesn't use index\n");
      //
  
  }else{
      //traverse index to get the tuple
      int startKey;
      int endKey;
      vector<SelCond> new_cond;
      bool containValueinWhere = false;
      //get a filtered filtered list of conditions and check that there are no conflicted conditions
      if(filterSelCond(cond, new_cond, startKey, endKey, containValueinWhere)){

          RC rc;
          IndexCursor cursor;
	  cursor.pid=-10;
          //locate the leaf-node index entry whose key value is larger than or equal to the startKey
          rc = tree.locate(startKey, cursor);
          

          if(rc < 0) return rc;
          // printCond(new_cond);
          //Read the (key, rid) pair at the cursor and move forward the cursor
          //TODO: when read till the end of the node, move on to the sibling node.
          while( tree.readForward(cursor, startKey, rid) == 0 && startKey <= endKey){

              
               //cases when we need to read the value
               //1: select *, select key,value, select value
               //2: value statement in where
              
              if(containValueinWhere == true || attr == 2 || attr == 3){
                  
                  if ((rc = rf.read(rid, key, value)) < 0) {
                      fprintf(stderr, "Error: while reading a tuple from table %s\n", table.c_str());
                      tree.close();
                      return rc;
                  }
                  bool rightTuple = true;
                   //a flag to check whether the tuple meets the condition
                  
                  //check value and NE conditions
                  for(unsigned i = 0; i< new_cond.size(); i++){
                      switch (new_cond[i].attr) {          
                          case 1:
                              diff = key - atoi(new_cond[i].value);
                              break;
                          case 2:
                            
                              diff = strcmp(value.c_str(), new_cond[i].value);
                              //cout <<"diff: "<<diff<<endl;
                              break;
                      }
                      
                      // skip the tuple if any condition is not met
                      switch (new_cond[i].comp) {
                          case SelCond::EQ:
                              if (diff != 0) rightTuple = false;
                              break;
                          case SelCond::NE:
                              if (diff == 0) rightTuple = false;
                              break;
                          case SelCond::GT:
                              if (diff <= 0) rightTuple = false;
                              break;
                          case SelCond::LT:
                              if (diff >= 0) rightTuple = false;    
                              break;
                          case SelCond::GE:
                               if (diff < 0) rightTuple = false;
                              break;
                          case SelCond::LE:
                              if (diff > 0) rightTuple = false;
                              break;
                      }
                  }
                  if(rightTuple){
                      // the condition is met for the tuple. 
                      // increase matching tuple counter
                     count++;
                      // print the tuple 
                      switch (attr) {
                          case 1:  // SELECT key
                              fprintf(stdout, "%d\n", key);
                              break;
                          case 2:  // SELECT value
                              fprintf(stdout, "%s\n", value.c_str());
                              break;
                          case 3:  // SELECT *
                             fprintf(stdout, "%d '%s'\n", key, value.c_str());
                             break;
                      }
                  }
                    
              }else{
                  // the condition is met for the tuple. 
                  // increase matching tuple counter
                  count++;
                  // print the tuple with key
                  if(attr == 1)
                       fprintf(stdout, "%d\n", startKey);
                  
              }
              
              
              
          }
      }
      //DEBUG
      //fprintf(stderr, "\n\nThis use index\n");
  }
   
  
  //DEBUG
        //fprintf(stdout, "%d\n", count);
  // print matching tuple count if "select count(*)"
  if (attr == 4) {
      fprintf(stdout, "%d\n", count);
  }
  rc = 0;

  // close the table file and return
  exit_select:
  rf.close();
  return rc;
}

RC SqlEngine::load(const string& table, const string& loadfile, bool index)
{
  //Example input: LOAD movie FROM 'movieData.del'
  //Index file: 'movieData.idx'
    
  //Open Load file
  fstream fp;
  fp.open (loadfile.c_str(), fstream::in);

  //if failed to open the file
  if(fp.fail())
  {
      fp.close();
      return RC_FILE_OPEN_FAILED;
  }
  
  //open or create a new record file
  RC rc;
  RecordFile newTable;
  
  if((rc = newTable.open(table + ".tbl", 'w')) < 0)
  {
      //open record file failed
      fp.close();
      newTable.close();
      return rc; 
  }
  
  //if index option is true
  BTreeIndex tree;
  if(index == true)
  {
      if( (rc = tree.open(table + ".idx", 'w')) < 0)
      {
          //open index file failed
          fp.close();
          newTable.close();
          tree.close();
          return rc;
      }
  }
  
  //read each line of tuples
  string line;
  int countLine =0;
  while(fp.good() && getline(fp,line))
  {
      int key;
      string value;
      RecordId rid;
      countLine++;
      //parse the line
      if( (rc = parseLoadLine(line, key, value)) < 0)
      {
          //if there's an error parsing the line
          fp.close();
          newTable.close();
          if(index)
          {
              tree.close();
          }
          return rc;
      }
      
      //append the data to the table file
      if( (rc = newTable.append(key,value,rid)) < 0)
      {
          //if there's an error inserting data to the table
          fp.close();
          newTable.close();
          if(index == true)
          {
              tree.close();
          }
          return rc;
      }
      
      //DEBUG
      //fprintf(stderr, "1\n");
      //if index option is on, insert data to BTree
      if(index)
      {
          //fprintf(stderr, "Insert %d tuple key %u value $s\n", countLine, key,value.c_str());
          if( (rc = tree.insert(key,rid)) < 0)
          {
              fp.close();
              newTable.close();
              tree.close();
              
              
              return rc;
          }
          
          
          //DEBUG
          //fprintf(stderr, "%d '%s'\n", key, value.c_str());
      }
      
      //DEBUG
      //fprintf(stderr, "2\n");
  }
  
  fp.close();
  newTable.close();
  if(index)
  {
      tree.close();
  }
  
  return 0;
}

RC SqlEngine::parseLoadLine(const string& line, int& key, string& value)
{
    const char *s;
    char        c;
    string::size_type loc;
    
    // ignore beginning white spaces
    c = *(s = line.c_str());
    while (c == ' ' || c == '\t') { c = *++s; }

    // get the integer key value
    key = atoi(s);

    // look for comma
    s = strchr(s, ',');
    if (s == NULL) { return RC_INVALID_FILE_FORMAT; }

    // ignore white spaces
    do { c = *++s; } while (c == ' ' || c == '\t');
    
    // if there is nothing left, set the value to empty string
    if (c == 0) { 
        value.erase();
        return 0;
    }

    // is the value field delimited by ' or "?
    if (c == '\'' || c == '"') {
        s++;
    } else {
        c = '\n';
    }

    // get the value string
    value.assign(s);
    loc = value.find(c, 0);
    if (loc != string::npos) { value.erase(loc); }

    return 0;
}
